package org.ccay.async.conduit;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.inject.Inject;

import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;

import org.ccay.async.IAsyncFramework;
import org.ccay.async.IMessageProcessor;
import org.ccay.async.AsyncMessage;
import org.ccay.core.exception.ApplicationException;
import org.ccay.core.log.ILogger;
import org.ccay.core.log.CcayLoggerFactory;

/**
 * 多线程异步通道。
 * 	
 * 
 * @since Dec 6, 2011
 */
public class ThreadsConduit extends AbstractAsyncConduit implements InitializingBean, DisposableBean {
	
	private final ILogger logger = CcayLoggerFactory.getLogger(ThreadsConduit.class);
	
	/** 一分钟对应的秒数 **/
	private static final int ONE_MINUTE = 60;

	/** 对象锁 **/
	private final Object poolSizeMonitor = new Object();

	/** 核心线程池的大小 **/
	private int corePoolSize = 1;

	/** 线程池的最大线程数 **/
	private int maxPoolSize = Integer.MAX_VALUE;

	/** 线程空闲时的存活时间 **/
	private int keepAliveSeconds = ONE_MINUTE;

	/** 任务队列的容量 **/
	private int queueCapacity = Integer.MAX_VALUE;

	/** 多线程认为执行器 **/
	private ThreadPoolExecutor threadPoolExecutor;
	
	/**
	 * @return the threadPoolExecutor
	 */
	public ThreadPoolExecutor getThreadPoolExecutor() {
		return threadPoolExecutor;
	}

	/**
	 * @param threadPoolExecutor the threadPoolExecutor to set
	 */
	public void setThreadPoolExecutor(ThreadPoolExecutor threadPoolExecutor) {
		this.threadPoolExecutor = threadPoolExecutor;
	}

	@Inject
	private IAsyncFramework asyncFramework;

	/**
	 * 设置核心线程池的大小
	 * 	
	 * 
	 * @since Dec 6, 2011
	 * @param corePoolSize
	 */
	public void setCorePoolSize(int corePoolSize) {
		synchronized (this.poolSizeMonitor) {
			this.corePoolSize = corePoolSize;
			if (this.threadPoolExecutor != null) {
				this.threadPoolExecutor.setCorePoolSize(corePoolSize);
			}
		}
	}

	/**
	 * 获取核心线程池的大小
	 * 	
	 * 
	 * @since Dec 6, 2011
	 * @return
	 */
	public int getCorePoolSize() {
		synchronized (this.poolSizeMonitor) {
			return this.corePoolSize;
		}
	}

	/**
	 * 设置线程池的最大线程数
	 * 	
	 * 
	 * @since Dec 6, 2011
	 * @param maxPoolSize
	 */
	public void setMaxPoolSize(int maxPoolSize) {
		synchronized (this.poolSizeMonitor) {
			this.maxPoolSize = maxPoolSize;
			if (this.threadPoolExecutor != null) {
				this.threadPoolExecutor.setMaximumPoolSize(maxPoolSize);
			}
		}
	}

	/**
	 * 获取线程池的最大线程数。
	 * 	
	 * 
	 * @since Dec 6, 2011
	 * @return
	 */
	public int getMaxPoolSize() {
		synchronized (this.poolSizeMonitor) {
			return this.maxPoolSize;
		}
	}

	/**
	 * 设置线程空闲时的存活时间，单位为秒。
	 * 	
	 * 
	 * @since Dec 6, 2011
	 * @param keepAliveSeconds
	 */
	public void setKeepAliveSeconds(int keepAliveSeconds) {
		synchronized (this.poolSizeMonitor) {
			this.keepAliveSeconds = keepAliveSeconds;
			if (this.threadPoolExecutor != null) {
				this.threadPoolExecutor.setKeepAliveTime(keepAliveSeconds, TimeUnit.SECONDS);
			}
		}
	}

	/**
	 * 获取线程空闲存活的时间。
	 * 	
	 * 
	 * @since Dec 6, 2011
	 * @return
	 */
	public int getKeepAliveSeconds() {
		synchronized (this.poolSizeMonitor) {
			return this.keepAliveSeconds;
		}
	}

	public void setQueueCapacity(int queueCapacity) {
		this.queueCapacity = queueCapacity;
	}

	/* (non-Javadoc)
	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
	 */
	public void afterPropertiesSet() throws ApplicationException {
		Assert.notNull(asyncFramework);
		
		// 初始化多线程通道
		initialize();
	}

	/**
	 * 初始化多线程任务执行器。
	 * 
	 * 
	 * @since Dec 6, 2011
	 */
	public void initialize() {
		BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);
		this.threadPoolExecutor = createThreadPoolExecutor(queue);
	}
	
	/**
	 * 创建一个Executor
	 * 	
	 * 
	 * @since Dec 16, 2011
	 * @return
	 */
	protected ThreadPoolExecutor createThreadPoolExecutor(BlockingQueue<Runnable> queue) {
		return new ThreadPoolExecutor(
				this.getCorePoolSize(),this.getMaxPoolSize(),this.getKeepAliveSeconds()
				, TimeUnit.SECONDS,
				queue, 
				new RejectedExecutionHandler(){

					/* (non-Javadoc)
					 * @see java.util.concurrent.RejectedExecutionHandler#rejectedExecution(java.lang.Runnable, java.util.concurrent.ThreadPoolExecutor)
					 */
					public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
						AsyncMessage message = ((NamedRunnable)r).getMessage();
						logger.warn("Threads conduit could not process message("+
								message.getType()+"), maybe threads exceed max threads("+getMaxPoolSize()
								+") and task queue is full!");
					}
					
				});
	}

	/**
	 * 创建阻塞队列。
	 * 	
	 * 
	 * @since Dec 6, 2011
	 * @param queueCapacity
	 * @return
	 */
	protected BlockingQueue<Runnable> createQueue(int queueCapacity) {
		if (queueCapacity > 0) {
			return new LinkedBlockingQueue<Runnable>(queueCapacity);
		}
		else {
			return new SynchronousQueue<Runnable>();
		}
	}

	/* (non-Javadoc)
	 * @see org.ccay.core.async.AsyncConduit#handle(org.ccay.core.async.Message)
	 */
	public void handle(final AsyncMessage message) {
		this.threadPoolExecutor.execute(new NamedRunnable(message));
	}

	/* (non-Javadoc)
	 * @see org.springframework.beans.factory.DisposableBean#destroy()
	 */
	public void destroy() throws ApplicationException {
		this.threadPoolExecutor.shutdown();
	}
	
	class NamedRunnable implements Runnable {
		
		/** 待处理的详细 **/
		private AsyncMessage message;
		
		/**
		 * 构造函数
		 */
		public NamedRunnable(AsyncMessage message) {
			this.message = message;
		}
		
		/**
		 * 获取待处理的Message	
		 * 
		 * @since Dec 21, 2011
		 * @return
		 */
		public AsyncMessage getMessage() {
			return message;
		}

		/* (non-Javadoc)
		 * @see java.lang.Runnable#run()
		 */
		public void run() {
			// 获取消息处理器
			IMessageProcessor processor = asyncFramework.getMessageProcessor(message);
			if(processor != null) {
				try {
					processor.process(message);
				} catch (ApplicationException e) {
					logger.error(e);
				}  catch(RuntimeException e){
					logger.error(e);
				} catch(Error e){
					//记录日志以便定位问题
					logger.fatal(e);
					throw e;
				}
			} else {
				logger.warn("Could not find message processor for message: "+message.getType());
			}
		}
		
	}

}